package org.redis.cluster.redis.transport.codec.server;

import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.RedisProtocolType;
import org.redis.cluster.transport.ByteBuffer;
import org.redis.cluster.transport.protocol.ProtocolOutput;
import org.redis.cluster.transport.session.Session;
import org.redis.cluster.transport.support.AbstractReplayingProtocolDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RedisServerProtocolDecoder extends AbstractReplayingProtocolDecoder {
	
	private final static byte LINE_END_BYTE='\r';
	
	private final static byte PARAMETER_START_BYTE='*';
	
	private final static byte PARAMETER_LENGTH_BYTE='$';
	
	private final static String COMMAND_CONTEXT="_COMMAND";
	
	private final static String COMMAND_PARAMETER_INDEX="_COMMAND_PARAMETER_INDEX";
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private RedisServerInlineProtocolDecoder inlineDecoder;
	
	public RedisServerProtocolDecoder() {
		this.inlineDecoder=new RedisServerInlineProtocolDecoder();
	}

	@Override
	protected void doDecode(ByteBuffer buffer, Session session, ProtocolOutput output) {
		if(!buffer.isReadable()) {
			return;
		}
		
		RedisCommand command=getCommand(session);
		if(command.getProtocolType()==null) {
			if(buffer.getByte(0)==PARAMETER_START_BYTE) {
				command.setProtocolType(RedisProtocolType.MULTIBULK);
			}
			else {
				command.setProtocolType(RedisProtocolType.INLINE);
			}
		}
		
		if(command.getProtocolType().equals(RedisProtocolType.MULTIBULK)) {
			doDecodeMultibulk(command, buffer, session, output);
		}
		else {
			inlineDecoder.decode(command, buffer, session, output);
			clean(session);
		}
	}
	
	protected void doDecodeMultibulk(RedisCommand command,ByteBuffer buffer,Session session,ProtocolOutput output) {
		if(command.getParameterCount()==0 && buffer.readByte()==PARAMETER_START_BYTE) {
			int parameterCount=readInt(buffer);
			checkPosition(session, buffer);
			
			command.setParameterCount(parameterCount);
			command.setParameters(new byte[parameterCount][]);
			
			if(logger.isDebugEnabled()) {
				logger.debug("接收到参数数量为 {} 的命令",parameterCount);
			}
		}
		
		for(int i=getParameterIndex(session),len=command.getParameterCount();i<len;i++) {
			buffer.skip(1);
			
			int length=readInt(buffer);
			
			if(logger.isDebugEnabled()) {
				logger.debug("接收到参数长度为 {} 的命令",length);
			}
			
			byte parameter[]=new byte[length];
			
			buffer.readBytes(parameter);
			buffer.skip(2);
			
			setParameterIndex(session, i+1);
			checkPosition(session, buffer);
			
			command.getParameters()[i]=parameter;
		}
		
		clean(session);
		
		command.setSource(buffer.array());
		output.out(session, command);
	}

	protected int readInt(ByteBuffer buffer) {
		int result=0;
		byte temp;
		
		while((temp=buffer.readByte())!=LINE_END_BYTE) {
			result=(result*10)+(temp-'0');
		}
		
		buffer.skip(1);
		return result;
	}
	
	private void clean(Session session) {
		session.getContext().remove(COMMAND_CONTEXT);
		session.getContext().remove(COMMAND_PARAMETER_INDEX);
	}
	
	private RedisCommand getCommand(Session session) {
		RedisCommand command=session.getContext().get(COMMAND_CONTEXT);
		if(command==null) {
			command=new RedisCommand();
			session.getContext().put(COMMAND_CONTEXT, command);
		}
		
		return command;
	}
	
	protected int getParameterIndex(Session session) {
		Integer index=session.getContext().get(COMMAND_PARAMETER_INDEX);
		if(index==null) {
			return 0;
		}
		
		return index;
	}
	
	protected boolean setParameterIndex(Session session,int index) {
		return session.getContext().put(COMMAND_PARAMETER_INDEX, index);
	}
}